package vip.xiaonuo.transaction;

import cn.hutool.core.exceptions.ExceptionUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.redisson.api.RBucket;
import org.redisson.api.RList;
import org.redisson.api.RedissonClient;
import org.springframework.messaging.Message;
import vip.xiaonuo.mapper.ThreeSidesMapper;
import vip.xiaonuo.constant.ThreeSidesConstant;

import javax.annotation.Resource;
import java.util.Objects;

@RocketMQTransactionListener
@Slf4j
public class ThreeSidesTransactionListener implements RocketMQLocalTransactionListener {

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private ThreeSidesMapper threeSidesMapper;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            String batchId = Objects.requireNonNull(msg.getHeaders().get(ThreeSidesConstant.BATCH_ID)).toString();
            RList<Long> list = redissonClient.getList(batchId);
            threeSidesMapper.updateBatchPushFlag(list, (String) msg.getHeaders().get(ThreeSidesConstant.TABLE));
            list.delete();
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("表【{}】推送任务执行本地消息发生错误【{}】", msg.getHeaders().get(ThreeSidesConstant.TABLE), ExceptionUtil.getMessage(e));
        } finally {
            // 删除key
            RBucket<String> vehicle_blacklist_tx = redissonClient.getBucket((String) msg.getHeaders().get(ThreeSidesConstant.TASK_LOCK_KEY));
            vehicle_blacklist_tx.delete();
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        try {
            String batchId = Objects.requireNonNull(msg.getHeaders().get(ThreeSidesConstant.BATCH_ID)).toString();
            RList<Long> list = redissonClient.getList(batchId);
            Long count = threeSidesMapper.selectCountPushFlag(list, (String) msg.getHeaders().get(ThreeSidesConstant.TABLE));
            if (count == 0) {
                return RocketMQLocalTransactionState.COMMIT;
            } else {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e) {
            log.error("表【{}】推送任务回查发生错误【{}】", msg.getHeaders().get(ThreeSidesConstant.TABLE), ExceptionUtil.getMessage(e));
        } finally {
            // 删除key
            RBucket<String> vehicle_blacklist_tx = redissonClient.getBucket((String) msg.getHeaders().get(ThreeSidesConstant.TASK_LOCK_KEY));
            vehicle_blacklist_tx.delete();
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}
